package ct.consumer.bean;

import ct.common.bean.Consumer;
import ct.consumer.dao.*;
import ct.common.constant.Names;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.util.Arrays;
import java.util.Properties;

/**
 * @Author Jeremy Zheng
 * @Date 2021/2/20 16:41
 * @Version 1.0
 * 通话日志的消费者对象
 */
public class CallLogConsumer implements Consumer {
    //消费数据
    @Override
    public void consume() {
        try {
            //创建配置对象
            Properties prop = new Properties();
            prop.load(Thread.currentThread().getContextClassLoader()
                    .getResourceAsStream("consumer.properties"));
            //kafka消费者获取flume采集的数据
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
            //关注主题
            consumer.subscribe(Arrays.asList(Names.TOPIC.getValue()));

            //Hbase数据访问对象
            HbaseDao dao =new HbaseDao();
            dao.init();

            //消费数据
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(100);
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    System.out.println(consumerRecord.value());
                    //将数据插入到Hbase
                    dao.insertData(consumerRecord.value());
                    //CallLog log=new CallLog(consumerRecord.value());
                    //dao.insertData(log);
                }
            }

        }catch (Exception e){
            e.printStackTrace();
        }

    }

    //关闭资源
    @Override
    public void close() throws IOException {

    }

}
